package com.scala.CopeWithData

import java.util.UUID

import com.java.config.IpConfig
import com.java.util.{InitialUtil, PatternUtil}
import com.scala.CaseClass.scoreInfo
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode, SparkSession}
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service

/**
 * @Author: ZhangJin
 * @Date: 2020/9/20 21:53
 */
// TODO:   处理附近数据
@Service("CopeWithDataForNearBy")
class CopeWithDataForNearBy @Autowired()(val session: SparkSession,val ipConfig:IpConfig) extends Serializable{

  // TODO:   附近竞争对手的数据处理
  def CopeNearByData(fileName: String,SQLforTable: String) {
    import session.implicits._
    val hc: SQLContext = session.sqlContext
    // TODO:   去除null、NaN
    val DF: DataFrame = hc.sql("select * from default." + fileName).as[scoreInfo].rdd.
      filter(row => {
        PatternUtil.MatchPatternForNearByStore(row.consumerByAvg, row.rateStore)
      }).map(each => {
      // TODO:  增加UUid
      val UUid =UUID.randomUUID().toString
      val location = each.Location
      val LocationOfLatAndLong = InitialUtil.getcoorder(each.Location)
      val rateStore = each.rateStore
      val consumerByAvg = each.consumerByAvg
      val numberCommit = each.numberCommit
      val Url = each.Url
      val Type = each.Type
      val nameOfStore = each.nameOfStore
      println("位于" + each.Location + "的" +each.nameOfStore + "的经纬度为: " + LocationOfLatAndLong)
      scoreInfo(UUid,nameOfStore, Url, location, consumerByAvg, rateStore, numberCommit, Type, LocationOfLatAndLong)
    }).toDF().na.drop().where("type <> '' ").where("ratestore  <> '' ").where("consumerbyavg  <> '' ").where("locationoflatandlong <> '' ")
    DF.show(20)
    // TODO:   处理后的数据进行持久化
    hc.sql("CREATE DATABASE IF NOT EXISTS CopeWithData")
    hc.sql("use CopeWithData")
    /* codeLaterRentInfoData"+ fileName */
    DF.write.mode(SaveMode.Overwrite).parquet("hdfs://"+ipConfig.getIp+":"+ipConfig.getPort+"/usr/data/codeLaterNearByInfoData"+ fileName +".parquet")
    hc.sql("create table if not exists codeLaterNearByInfoData"+ fileName + SQLforTable + "stored as parquet")
    hc.sql("load data inpath 'hdfs://" + ipConfig.getIp+":"+ipConfig.getPort+ "/usr/data/codeLaterNearByInfoData"+ fileName +".parquet' overwrite into table codeLaterNearByInfoData" + fileName)
  }

}
